package mr.stocks.question3;

import java.io.IOException;

import mr.stocks.avro.model.StockPriceByYearKeyAvro;
import mr.stocks.avro.model.StockPriceByYearReportAvro;
import mr.stocks.avro.model.StockPriceByYearValueAvro;
import mr.stocks.avro.model.StocksReportAvro;
import mr.support.ConfHelper;

import org.apache.avro.mapreduce.AvroJob;
import org.apache.avro.mapreduce.AvroKeyInputFormat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class StocksAnnualAverageJob extends Configured implements Tool {

	private static Logger LOG = LoggerFactory.getLogger(StocksAnnualAverageJob.class);

	@Override
	public int run(String[] arg0) throws Exception {
		Job job = Job.getInstance(getConf(), this.getClass().getName());
		job.setJarByClass(getClass());
		configureJob(job);
		return job.waitForCompletion(true) ? 0 : 1;
	}

	private void configureJob(Job job) throws IOException {
		Configuration jobConf = job.getConfiguration();
		ConfHelper confHelper = new ConfHelper(jobConf);
		
		job.setInputFormatClass(AvroKeyInputFormat.class);
		AvroKeyInputFormat.addInputPath(job, confHelper.getInput());
		AvroJob.setInputKeySchema(job, StocksReportAvro.SCHEMA$);
		
		Path outPath = confHelper.getOutputPath();
		TextOutputFormat.setOutputPath(job, outPath);
		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(NullWritable.class);

		job.setMapperClass(StockPriceByYearAvroMapper.class);
		job.setReducerClass(StockPriceByYearAvroReducer.class);
		job.setCombinerClass(StockPriceByYearAvroCombiner.class);
		
		job.setMapOutputKeyClass(Text.class);
		AvroJob.setMapOutputValueSchema(job, StockPriceByYearReportAvro.SCHEMA$);
	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		int exitCode = ToolRunner.run(new StocksAnnualAverageJob(), args);
		System.exit(exitCode);
	}

}
